use chrono::{Duration, NaiveDateTime};
use crate::physical_plan::PhysicalPlanTrait;
use chrono::{Duration as ChronoDuration, TimeZone};
use crate::run_plan::{HandResult, HandResultOk, HandResultFail};
use std::cmp::Ordering;
use std::sync::Arc;
use std::collections::{HashMap, VecDeque};

pub mod json;

pub trait DataFrame: Sync+Send{
    fn get_value(&self, key: &ElementValue)->HandResult;
}

#[derive(Clone)]
pub enum Data{
    DataFrame(Arc<dyn DataFrame>),
    DataFrameList(Vec<Arc<dyn DataFrame>>),
}

//配置也页面值，不指定值的类型，以期更通用
#[derive(Debug,Clone)]
pub struct ConfValue{
    pub value: String,
    rv_u64: Option<RealValue>,
    rv_boolean: Option<RealValue>,
    rv_datetime: Option<RealValue>,
    rv_null: Option<RealValue>,
    rv_utf8: Option<RealValue>,
    rv_i64: Option<RealValue>,
    rv_f64: Option<RealValue>,
    rv_range: Option<RealValue>,
    rv_durataion: Option<RealValue>,
}

impl ConfValue{
    fn _to_u64(&mut self){
        if let Ok(t) = self.value.parse::<u64>(){
            self.rv_u64 = Some(RealValue::U64(t));
        }
    }
    fn _to_i64(&mut self){
        if let Ok(t) = self.value.parse::<i64>(){
            self.rv_u64 = Some(RealValue::I64(t));
        }
    }
    fn _to_f64(&mut self){
        if let Ok(t) = self.value.parse::<f64>(){
            self.rv_f64 = Some(RealValue::F64(t));
        }
    }
    fn _to_boolean(&mut self){
        if self.value == "yes" ||
            self.value == "YES" ||
            self.value == "ok" ||
            self.value == "OK" ||
            self.value == "true" ||
            self.value == "TRUE"{
            self.rv_boolean = Some(RealValue::Boolean(true));
        }else if self.value == "no" ||
            self.value == "NO" ||
            self.value == "fail" ||
            self.value == "FAIL" ||
            self.value == "FALSE"{
            self.rv_boolean = Some(RealValue::Boolean(false));
        }
    }

    pub fn new(value: String)->ConfValue{
        let mut tmp = ConfValue{
            value: value,
            rv_u64: None,
            rv_boolean: None,
            rv_datetime: None,
            rv_null: None,
            rv_utf8: None,
            rv_i64: None,
            rv_f64: None,
            rv_range: None,
            rv_durataion: None,
        };
        tmp._to_u64();
        tmp._to_boolean();
        tmp
    }

    pub fn to_datetime(&self)->HandResult{
        if let Some(ref t) = self.rv_datetime{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to DateTime failed"));
        }
    }
    pub fn to_u64(&self)->HandResult{
        if let Some(ref t) = self.rv_u64{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to UInt64 failed"));
        }
    }
    pub fn to_i64(&self)->HandResult{
        if let Some(ref t) = self.rv_u64{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to Int64 failed"));
        }
    }
    pub fn to_f64(&self)->HandResult{
        if let Some(ref t) = self.rv_f64{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to F64 failed"));
        }
    }
    pub fn to_boolean(&self)->HandResult{
        if let Some(ref t) = self.rv_boolean{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to Boolean failed"));
        }
    }
    pub fn to_null(&self) -> HandResult{
        if let Some(ref t) = self.rv_null{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to Null failed"));
        }
    }
    pub fn to_utf8(&self) -> HandResult{
        if let Some(ref t) = self.rv_utf8{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to Utf8 failed"));
        }
    }
    pub fn to_range(&self) -> HandResult{
        if let Some(ref t) = self.rv_range{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to Range failed"));
        }
    }
    pub fn to_duration(&self) -> HandResult{
        if let Some(ref t) = self.rv_durataion{
            return HandResultOk::new(t.clone()).into();
        }else{
            return HandResult::new_fail(format!("ConfValue to Duration failed"));
        }
    }

    pub fn lt(&self, right: &RealValue)->HandResult{
        return self.change(right)?.value.lt(right);
    }
    pub fn _lt(left: &RealValue, right: &ConfValue)->HandResult{
        return left.lt(&right.change(left)?.value);
    }
    pub fn le(&self, right: &RealValue)->HandResult{
        return self.change(right)?.value.le(right);
    }
    pub fn _le(left: &RealValue, right: &ConfValue)->HandResult{
        return left.le(&right.change(left)?.value);
    }
    pub fn gt(&self, right: &RealValue)->HandResult{
        return self.change(right)?.value.gt(right);
    }
    pub fn _gt(left: &RealValue, right: &ConfValue)->HandResult{
        return left.gt(&right.change(left)?.value);
    }
    pub fn ge(&self, right: &RealValue)->HandResult{
        return self.change(right)?.value.ge(right);
    }
    pub fn _ge(left: &RealValue, right: &ConfValue)->HandResult{
        return left.ge(&right.change(left)?.value);
    }
    pub fn eq(&self, right: &RealValue)->HandResult{
        return self.change(right)?.value.eq(right);
    }
    pub fn _eq(left: &RealValue, right: &ConfValue)->HandResult{
        return left.eq(&right.change(left)?.value);
    }
    pub fn neq(&self, right: &RealValue)->HandResult{
        return self.change(right)?.value.neq(right);
    }
    pub fn _neq(left: &RealValue, right: &ConfValue)->HandResult{
        return left.neq(&right.change(left)?.value);
    }
    fn change(&self, other: &RealValue)->HandResult{
        let left = match other{
            RealValue::U64(_) => { self.to_u64()? },
            RealValue::I64(_) => { self.to_i64()? },
            RealValue::F64(_) => { self.to_f64()? },
            RealValue::Null => { self.to_null()? },
            RealValue::Boolean(_) => { self.to_boolean()? },
            RealValue::Duration(_) => { self.to_duration()? },
            RealValue::Utf8(_) => { self.to_utf8()? },
            RealValue::Range {left,right} => { self.to_range()? },
            RealValue::DateTime(_) => { self.to_datetime()? }
        };
        return HandResultOk::new(left.value).into();
    }
}

//有具体类型的值，需要严格值类型匹配
#[derive(Clone,Debug)]
pub enum RealValue{
    Null,            //用于某些get情况下的不存在或者某些执行函数不需要返回值
    Boolean(bool),
    Utf8(String),
    U64(u64),
    I64(i64),
    F64(f64),
    Range{
        left: u64,
        right: u64,

    },
    DateTime(NaiveDateTime),
    Duration(ChronoDuration),
}

impl RealValue{
    pub fn gt(&self, right: &RealValue) ->HandResult{
        match self{
            RealValue::Null => {return HandResult::new_fail(format!("gt left is Null"));},
            RealValue::U64(left) => {
                if let RealValue::U64(right) = right {
                    return HandResultOk::new(RealValue::Boolean(left > right)).into();
                }else{
                    return HandResult::new_fail(format!("gt left is UInt64, but right not"));
                }
            },
            RealValue::F64(left) => {
                if let RealValue::F64(right) = right {
                    return HandResultOk::new(RealValue::Boolean(left > right)).into();
                }else{
                    return HandResult::new_fail(format!("gt left is F64, but right not"));
                }
            },
            RealValue::Boolean(_) => {return HandResult::new_fail(format!("gt left is Boolean"));},
            RealValue::Range{left,right} => {return HandResult::new_fail(format!("gt left is Range"));},
            RealValue::DateTime(left_value) => {
                if let RealValue::DateTime(right_value) = right {
                    return HandResultOk::new(RealValue::Boolean(left_value > right_value)).into();
                }else{
                    return HandResult::new_fail(format!("gt left is DateTime, but right not"));
                }
            },
            RealValue::Utf8(_) => {return HandResult::new_fail(format!("gt left is Utf8"));},
            RealValue::Duration(_) => {return HandResult::new_fail(format!("gt left is Duration"));},
            RealValue::I64(left_value) => {
                if let RealValue::I64(right_value) = right {
                    return HandResultOk::new(RealValue::Boolean(left_value > right_value)).into();
                }else{
                    return HandResult::new_fail(format!("gt left is Int64, but right not"));
                }
            },
        }
    }
    pub fn ge(&self, right: &RealValue) ->HandResult{
        match self{
            RealValue::Null => {return HandResult::new_fail(format!("ge left is Null"));},
            RealValue::U64(left) => {
                if let RealValue::U64(right) = right {
                    return HandResultOk::new(RealValue::Boolean(left >= right)).into();
                }else{
                    return HandResult::new_fail(format!("ge left is UInt64, but right not"));
                }
            },
            RealValue::F64(left) => {
                if let RealValue::F64(right) = right {
                    return HandResultOk::new(RealValue::Boolean(left >= right)).into();
                }else{
                    return HandResult::new_fail(format!("ge left is F64, but right not"));
                }
            },
            RealValue::Boolean(_) => {return HandResult::new_fail(format!("ge left is Boolean"));},
            RealValue::Range{left,right} => {return HandResult::new_fail(format!("gt left is Range"));},
            RealValue::DateTime(left_value) => {
                if let RealValue::DateTime(right_value) = right {
                    return HandResultOk::new(RealValue::Boolean(left_value >= right_value)).into();
                }else{
                    return HandResult::new_fail(format!("ge left is DateTime, but right not"));
                }
            },
            RealValue::Utf8(_) => {return HandResult::new_fail(format!("ge left is Utf8"));},
            RealValue::Duration(_) => {return HandResult::new_fail(format!("ge left is Duration"));},
            RealValue::I64(left_value) => {
                if let RealValue::I64(right_value) = right {
                    return HandResultOk::new(RealValue::Boolean(left_value >= right_value)).into();
                }else{
                    return HandResult::new_fail(format!("ge left is Int64, but right not"));
                }
            },
        }
    }
    pub fn lt(&self, right: &RealValue) ->HandResult{
        match self{
            RealValue::Null => {return HandResult::new_fail(format!("lt left is Null"));},
            RealValue::U64(left) => {
                if let RealValue::U64(right) = right {
                    return HandResultOk::new(RealValue::Boolean(left < right)).into();
                }else{
                    return HandResult::new_fail(format!("lt left is UInt64, but right not"));
                }
            },
            RealValue::F64(left) => {
                if let RealValue::F64(right) = right {
                    return HandResultOk::new(RealValue::Boolean(left < right)).into();
                }else{
                    return HandResult::new_fail(format!("lt left is F64, but right not"));
                }
            },
            RealValue::Boolean(_) => {return HandResult::new_fail(format!("lt left is Boolean"));},
            RealValue::Range{left,right} => {return HandResult::new_fail(format!("gt left is Range"));},
            RealValue::DateTime(left_value) => {
                if let RealValue::DateTime(right_value) = right {
                    return HandResultOk::new(RealValue::Boolean(left_value < right_value)).into();
                }else{
                    return HandResult::new_fail(format!("lt left is DateTime, but right not"));
                }
            },
            RealValue::Utf8(_) => {return HandResult::new_fail(format!("lt left is Utf8"));},
            RealValue::Duration(_) => {return HandResult::new_fail(format!("lt left is Duration"));},
            RealValue::I64(left_value) => {
                if let RealValue::I64(right_value) = right {
                    return HandResultOk::new(RealValue::Boolean(left_value < right_value)).into();
                }else{
                    return HandResult::new_fail(format!("lt left is Int64, but right not"));
                }
            },
        }
    }
    pub fn le(&self, other: &RealValue)->HandResult{
        match self{
            RealValue::Null => {return HandResult::new_fail(format!("le left is Null"));},
            RealValue::Boolean(_) => {return HandResult::new_fail(format!("le left is Boolean"));},
            RealValue::Range{left,right} => {return HandResult::new_fail(format!("gt left is Range"));},
            RealValue::Utf8(_) => {return HandResult::new_fail(format!("le left is Utf8"));},
            RealValue::Duration(_) => {return HandResult::new_fail(format!("le left is Duration"));},
            RealValue::U64(left) => {
                if let RealValue::U64(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left <= right)).into();
                }else{
                    return HandResult::new_fail(format!("le left is UInt64, but right not"));
                }
            },
            RealValue::F64(left) => {
                if let RealValue::F64(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left <= right)).into();
                }else{
                    return HandResult::new_fail(format!("le left is F64, but right not"));
                }
            },
            RealValue::DateTime(left_value) => {
                if let RealValue::DateTime(right_value) = other{
                    return HandResultOk::new(RealValue::Boolean(left_value <= right_value)).into();
                }else{
                    return HandResult::new_fail(format!("le left is DateTime, but right not"));
                }
            },
            RealValue::I64(left_value) => {
                if let RealValue::I64(right_value) = other{
                    return HandResultOk::new(RealValue::Boolean(left_value <= right_value)).into();
                }else{
                    return HandResult::new_fail(format!("le left is Int64, but right not"));
                }
            },
        }
    }
    pub fn eq(&self, other: &RealValue)->HandResult{
        match self{
            RealValue::Null => {return HandResult::new_fail(format!("eq left is Null"));},
            RealValue::U64(left) => {
                if let RealValue::U64(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left == right)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is UInt64, but right not"));
                }
            },
            RealValue::F64(left) => {
                if let RealValue::F64(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left == right)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is F64, but right not"));
                }
            },
            RealValue::Boolean(left) => {
                if let RealValue::Boolean(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left == right)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is Boolean, but right not"));
                }
            },
            RealValue::Range{left,right} => {return HandResult::new_fail(format!("eq left is Range"));},
            RealValue::DateTime(left_value) => {
                if let RealValue::DateTime(right_value) = other{
                    return HandResultOk::new(RealValue::Boolean(left_value == right_value)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is DateTime, but right not"));
                }
            },
            RealValue::Utf8(left) => {
                if let RealValue::Utf8(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left == right)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is Utf8, but right not"));
                }
            },
            RealValue::Duration(left) => {
                if let RealValue::Duration(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left == right)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is Duration, but right not"));
                }
            },
            RealValue::I64(left_value) => {
                if let RealValue::I64(right_value) = other{
                    return HandResultOk::new(RealValue::Boolean(left_value == right_value)).into();
                }else{
                    return HandResult::new_fail(format!("eq left is Int64, but right not"));
                }
            },
        }
    }
    pub fn neq(&self, other: &RealValue)->HandResult{
        match self{
            RealValue::Null => {return HandResult::new_fail(format!("neq left is Null"));},
            RealValue::U64(left) => {
                if let RealValue::U64(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left != right)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is UInt64, but right not"));
                }
            },
            RealValue::F64(left) => {
                if let RealValue::F64(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left != right)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is F64, but right not"));
                }
            },
            RealValue::Boolean(left) => {
                if let RealValue::Boolean(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left != right)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is Boolean, but right not"));
                }
            },
            RealValue::Range{left,right} => {return HandResult::new_fail(format!("neq left is Range"));},
            RealValue::DateTime(left_value) => {
                if let RealValue::DateTime(right_value) = other{
                    return HandResultOk::new(RealValue::Boolean(left_value != right_value)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is DateTime, but right not"));
                }
            },
            RealValue::Utf8(left) => {
                if let RealValue::Utf8(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left != right)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is Utf8, but right not"));
                }
            },
            RealValue::Duration(left) => {
                if let RealValue::Duration(right) = other{
                    return HandResultOk::new(RealValue::Boolean(left != right)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is Duration, but right not"));
                }
            },
            RealValue::I64(left_value) => {
                if let RealValue::I64(right_value) = other{
                    return HandResultOk::new(RealValue::Boolean(left_value != right_value)).into();
                }else{
                    return HandResult::new_fail(format!("neq left is Int64, but right not"));
                }
            },
        }
    }
}

//另外一个函数返回的值
pub struct FuncValue{
     pub func_arg: Box<dyn PhysicalPlanTrait>,
}

//另外一个元素的值
#[derive(Clone,Debug,Eq,PartialEq,Hash)]
pub struct ElementValue{
    pub element: Vec<String>,
}

#[derive(Clone)]
pub struct DataInstance{
    data: Data, //data要在多个work node之间共享，所以不支持修改，修改记录放在metadata里面。
    pub metadata: MetaData,
}

impl DataInstance{
    pub fn new(data: Data)->DataInstance{
        return DataInstance{
            data: data,
            metadata: MetaData::new(),
        };
    }
}

#[derive(Clone)]
pub struct MetaData{
    high_light: HashMap<String,(usize, usize)>, //offset,length
    /*
    记录不同event_flow对事件的不同操作，目前主要是删除和重置值的记录。
    但是要注意一个问题：如果这里已经删了元素 a.b
    那么后续对比如a.b.c和a.b.e等元素的操作都失效
    */
    change_record: HashMap<ElementValue, ChangeRecord>,
    /*
    记录运行时错误，比如对非数字类型进行数字相关的操作或者判断等。
    该错误随着事件走，最后可以统一用一个节点来进行调试输出等
    hashmap的key分别是(节点名称，filter/func名称)
    */
    run_err: HashMap<(String, String), HandResultFail>,
    event_flow_path: VecDeque<String>,
}

impl MetaData{
    pub fn new()->MetaData{
        return MetaData{
            high_light: HashMap::new(),
            change_record: HashMap::new(),
            run_err: HashMap::new(),
            event_flow_path: VecDeque::new(),
        };
    }
    pub fn append_run_err(&mut self, node_name: &str, sub_name: &str, err: HandResultFail){
        self.run_err.insert((node_name.to_string(),sub_name.to_string()), err);
    }
    pub fn append_event_flow_path(&mut self, node_name: &str){
        self.event_flow_path.push_back(node_name.to_string());
    }
}

#[derive(Clone)]
pub enum ChangeRecord{
    DelKey,                              //删除k
    SetKey(RealValue),                      //给k设置值v
}

impl DataInstance{
    //filter 框架代码
    //func 框架代码
    pub fn get_value(&self, key: &ElementValue)->HandResult{
        if let Some(cache_data) = self.metadata.change_record.get(key){
            match cache_data{
                ChangeRecord::DelKey => {return HandResult::Err(HandResultFail::new(format!("{:?} deleted",key)));},
                ChangeRecord::SetKey(value) => {return HandResult::Ok(HandResultOk::new(value.clone()));},
            }
        }else{
            match &self.data{
                Data::DataFrame(data) => {
                    return data.get_value(key);
                },
                Data::DataFrameList(data_list) => {
                    return HandResult::Err(HandResultFail::new(format!("get_value not support DataFrameList")));
                },
            }
        }
    }
    pub fn set_value(&mut self, key: &ElementValue, value: &RealValue)->HandResult{
        self.metadata.change_record.insert(key.clone(), ChangeRecord::SetKey(value.clone()));
        return HandResult::Ok(HandResultOk::new(RealValue::Null));
    }
    pub fn del_key(&mut self, key: &ElementValue)->HandResult{
        self.metadata.change_record.insert(key.clone(), ChangeRecord::DelKey);
        return HandResult::Ok(HandResultOk::new(RealValue::Null));
    }
}
